package com.ruoyi.mqtt.service;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.ruoyi.mqtt.model.SensorData;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.time.Instant;
import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 测试类
 * 模拟大量传感器并发发送数据
 */
@Slf4j
public class MainTest {
    // 核心配置参数
    private static final int SENSOR_COUNT = 1;           // 使用单个传感器，减少客户端数量，避免多客户端竞争和资源消耗
    private static final int DATA_PER_SECOND = 10;     // 目标QPS仅作为参考，实际发送速率由系统处理能力决定
    private static final String BROKER_URL = "tcp://1.95.9.103:1883";
    private static final ObjectMapper objectMapper;      // 静态ObjectMapper，避免重复创建实例
    private static final AtomicLong totalSent = new AtomicLong(0);  // 原子计数器，线程安全的消息计数
    private static volatile boolean running = true;      // volatile标记，保证线程间的可见性，用于优雅停机
    private static final int BATCH_SIZE = 5;            // 批量发送大小，减少网络交互次数，提高吞吐量

    // 静态初始化ObjectMapper，只需初始化一次，提高性能
    static {
        objectMapper = new ObjectMapper();
        objectMapper.registerModule(new JavaTimeModule());
    }

    public static void main(String[] args) throws Exception {
        // 创建高效的线程池，使用CallerRunsPolicy避免任务丢失
        ExecutorService executorService = new ThreadPoolExecutor(
                SENSOR_COUNT,                    // 核心线程数与传感器数量一致
                SENSOR_COUNT,                    // 最大线程数固定，避免线程数波动
                60L,                            // 线程存活时间
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000), // 使用足够大的队列缓冲任务
                new ThreadPoolExecutor.CallerRunsPolicy() // 队列满时使用调用者线程执行任务
        );

        CountDownLatch latch = new CountDownLatch(SENSOR_COUNT);
        Random random = new Random();

        log.info("开始模拟 {} 个传感器发送数据，每秒目标 {} 条数据", SENSOR_COUNT, DATA_PER_SECOND);

        // 独立的统计线程，不影响主要业务逻辑的性能
        Thread statsThread = new Thread(() -> {
            long lastCount = 0;
            long startTime = System.currentTimeMillis();
            while (running) {
                try {
                    Thread.sleep(1000); // 每秒统计一次
                    long currentCount = totalSent.get();
                    long diff = currentCount - lastCount;
                    double qps = diff;
                    double avgQps = currentCount * 1000.0 / (System.currentTimeMillis() - startTime);
                    log.info("当前QPS: {}/s, 平均QPS: {}/s, 总计: {}",
                            String.format("%.2f", qps),
                            String.format("%.2f", avgQps),
                            currentCount);
                    lastCount = currentCount;
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        });
        statsThread.start();

        // 为每个传感器创建一个专用线程
        for (int i = 0; i < SENSOR_COUNT; i++) {
            final String sensorId = String.format("sensor-%04d", i);
            executorService.submit(() -> {
                MqttClient client = null;
                try {
                    // 使用MemoryPersistence避免磁盘IO
                    client = new MqttClient(BROKER_URL, sensorId, new MemoryPersistence());

                    // MQTT连接参数优化
                    MqttConnectOptions options = new MqttConnectOptions();
                    options.setCleanSession(true);      // 清除会话，减少状态维护开销
                    options.setMaxInflight(200);        // 提高单个客户端的并发处理能力
                    options.setConnectionTimeout(10);   // 较短的超时时间，快速失败
                    options.setKeepAliveInterval(30);   // 合理的心跳间隔
                    options.setAutomaticReconnect(true); // 自动重连，提高可靠性
                    client.connect(options);

                    // 主发送循环
                    while (running) {
                        try {
                            // 每次循环创建新的消息，确保时间戳不同
                            for (int j = 0; j < BATCH_SIZE; j++) {
                                SensorData data = new SensorData();
                                data.setSensorId(sensorId);
                                data.setValue(random.nextDouble() * 100);
                                data.setTimestamp(Instant.now().plusMillis(j + random.nextInt(10)));
                                data.setRoadName("Road " + random.nextInt(100));
                                data.setUnit("celsius");

                                String jsonData = objectMapper.writeValueAsString(data);
                                MqttMessage message = new MqttMessage(jsonData.getBytes());
                                message.setQos(0);

                                client.publish("sensors/data", message);
                                totalSent.incrementAndGet();
                            }

                            // 动态流控
                            if (client.getPendingDeliveryTokens().length > 500) {
                                Thread.sleep(5);
                            }
                        } catch (Exception e) {
                            log.warn("发送失败，重试中: {}", e.getMessage());
                            Thread.sleep(1000);  // 失败后等待更长时间

                            // 重新连接
                            if (!client.isConnected()) {
                                try {
                                    client.connect(options);
                                } catch (Exception reconnectError) {
                                    log.error("重连失败", reconnectError);
                                }
                            }
                        }
                    }
                } catch (Exception e) {
                    log.error("传感器 {} 发送失败", sensorId, e);
                } finally {
                    try {
                        if (client != null && client.isConnected()) {
                            client.disconnect();
                            client.close();
                        }
                    } catch (Exception e) {
                        log.error("关闭连接失败", e);
                    }
                    latch.countDown();
                }
            });
        }

        // 等待所有传感器启动
        latch.await();

        // 运行5分钟后关闭
        Thread.sleep(1 * 30 * 1000);
        running = false;
        statsThread.interrupt();
        statsThread.join();
        executorService.shutdownNow();
        executorService.awaitTermination(10, TimeUnit.SECONDS);

        log.info("测试完成，总共发送消息数: {}", totalSent.get());
    }
}
